package com.tc.vms.consumer;

import com.rabbitmq.client.ConsumerCancelledException;
import com.rabbitmq.client.MessageProperties;
import com.rabbitmq.client.QueueingConsumer;
import com.tc.vms.MessageWrap;
import com.tc.vms.VMSEventArgs;
import com.tc.vms.encoding.Message;
import com.tc.vms.exception.VmsClientException;
import com.tc.vms.iface.IContext;
import com.tc.vms.message.MessageConst;
import com.tc.vms.producer.VmsProducerRmqMultiQueueImpl;
import com.tc.vms.utils.ServiceThread;
import com.tc.vms.utils.Utils;
import com.tc.vms.utils.VmsThreadFactory;

import java.io.IOException;
import java.util.concurrent.Executors;

import static com.tc.vms.utils.Utils.MAX_DELAY_LEVEL;
import static com.tc.vms.utils.Utils.MIN_DELAY_LEVEL;

/**
 * Created by yonghua.zhang on 2016/6/14.
 */
public class VmsDLQConsumerRmqImpl extends VmsConsumerRmqImpl {

    public VmsDLQConsumerRmqImpl(VmsConsumerProxy vmsConsumerProxy) {
        super(vmsConsumerProxy);
    }

    @Override
    public synchronized boolean init(IContext context) throws VmsClientException {
        try {
            isQueueSharding = false;
            destAddresses = queueMetaData.getBrokerAddress();
            factory.setVirtualHost(queueMetaData.getVhost());
            factory.setUsername(queueMetaData.getUserName());
            factory.setPassword(queueMetaData.getUserPasswd());
            //factory.setAutomaticRecoveryEnabled(true);
            factory.setRequestedHeartbeat(Utils.RMQ_HEART_BEAT_INTERVAL);
            return true;
        } catch (Exception e) {
            throw new VmsClientException("VmsConsumerRmqImpl init failed.", e);
        }

    }

    @Override
    protected void commit(ConsumeStatus cs, MessageWrap message,
                          VmsRmqConsumerContext vmsRmqConsumerContext) {
        if (isConsumerRunning && channel != null && !subQoS.autoCommit()) {
            final long ackIdx = vmsRmqConsumerContext.getAckIdx();
            synchronized (channel) {
                try {
                    switch (cs) {
                        case CONSUME_SUCCESS:
                            channel.basicAck(ackIdx, false);
                            break;
                        case RECONSUME_LATER:
                            if (message.getReconsumeTimes() <= 0) {
                                MessageConst.setMessageReconsumeTimes(message, 0);
                                message.setDelayLevel(MIN_DELAY_LEVEL);
                            }

                            boolean isRetryMaxTimes = message.getReconsumeTimes() >= MAX_DELAY_LEVEL;
                            int consumeTimes = isRetryMaxTimes ? MAX_DELAY_LEVEL : message.getReconsumeTimes();
                            message.setDelayLevel(consumeTimes + 1);
                            MessageConst.setMessageReconsumeTimes(message, consumeTimes + 1);
                            Message.VMSMessage vmsMessage = VmsProducerRmqMultiQueueImpl.buildVmsMessage(message);
                            try {
                                if (!isRetryMaxTimes) {
                                    channel.basicPublish("", queueMetaData.getRealQueueNmOnBroker(),
                                            MessageProperties.PERSISTENT_TEXT_PLAIN, vmsMessage.toByteArray());
                                    channel.waitForConfirms(5000);
                                }
                                channel.basicAck(ackIdx, false);
                            } catch (Exception ex) {
                                LOGGER.warn("Sendback message exception and prefer to nack the message.", ex);
                                channel.basicNack(ackIdx, false, true);
                            }
                            break;
                        case CONSUME_DISCARD:
                        default:
                            channel.basicNack(ackIdx, false, false);
                            break;
                    }
                } catch (Exception ex) {
                    LOGGER.warn("COMMIT - Consumer ack exception. AckIdx = {}, channel is Active = {}", ackIdx,
                            channel.isOpen(), ex);
                    if (channel.isOpen()) {
                        try {
                            channel.basicRecover(true);
                        } catch (IOException e) {
                            LOGGER.error("COMMIT - Consumer basicRecover failed.", e);
                        }
                    }
                }
            }
        }
    }

    @Override
    protected void startConsume() throws VmsClientException {
        isConsumerRunning = true;
        try {
            queueingConsumer = new QueueingConsumer(channel);
            channel.basicConsume(queueMetaData.getRealQueueNmOnBroker(), subQoS.autoCommit(), instName, queueingConsumer);
            consumeService = Executors.newSingleThreadExecutor(new VmsThreadFactory("Vms_Rabbitmq_Consume_DLQ_Thread"));

            pullMessageService = new ServiceThread() {
                @Override
                public void run() {
                    while (isConsumerRunning) {
                        try {
                            final QueueingConsumer.Delivery delivery = queueingConsumer.nextDelivery();
                            while (vmsConsumerProxy.isSuspend()) {
                                LOGGER.debug("Consuming service is suspend by vmscenter");
                                Utils.sleep5s();
                            }
                            consumeService.submit(new Runnable() {
                                @Override
                                public void run() {
                                    ConsumeStatus cs = ConsumeStatus.CONSUME_DISCARD;
                                    VMSEventArgs consumeMessage = null;
                                    MessageWrap message = null;
                                    final VmsRmqConsumerContext vmsRmqConsumerContext = new VmsRmqConsumerContext(0,
                                            delivery.getEnvelope().getDeliveryTag());
                                    try {

                                        try {
                                            message = parseFromPayload(delivery);
                                            consumeMessage = new VMSEventArgs(true, message);
                                        } catch (Exception ex) {
                                            message = new MessageWrap(delivery.getBody());
                                            consumeMessage = new VMSEventArgs(false, message);
                                        }

                                        try {
                                            vmsRmqConsumerContext.setDelayLevelWhenReconsume(message.getDelayLevel());
                                            cs = callback.consumeMessage(vmsRmqConsumerContext, consumeMessage);
                                        } catch (Exception ex) {
                                            LOGGER.warn("invoke callback function failed.", ex);
                                        }
                                    } finally {
                                        if (isConsumerRunning) {
                                            commit(cs, message, vmsRmqConsumerContext);
                                        }
                                    }
                                }
                            });

                        } catch (InterruptedException e) {
                            LOGGER.warn("{} is interrupted.", getServiceName());
                            break;
                        } catch (ConsumerCancelledException cce) {
                            LOGGER.warn("Rabbitmq ConsumerCancelledException, restart consuming...", cce);
                            hasShutdownSignal = true;
                            break;
                        } catch (Exception otherEx) { //
                            LOGGER.warn("Rabbitmq pullMessageService error , sleep 3s ...", otherEx);
                            Utils.sleep3s();
                        }
                    }
                }

                @Override
                public String getServiceName() {
                    return "Rabbitmq_PullMessage_Service_" + System.currentTimeMillis();
                }
            };
            pullMessageService.start();
            LOGGER.debug("pullMessageService has been started. threadId=[{}]", Thread.currentThread().getId());
        } catch (Exception ex) {
            stopConsume();
            LOGGER.error("VmsConsumerRmqImpl startConsume failed.", ex);
            throw new VmsClientException("VmsConsumerRmqImpl startConsume failed.", ex);
        }
    }

    @Override
    protected void declareDelayQueues() throws VmsClientException {
        String qName = queueMetaData.getRealQueueNmOnBroker();
        try {
            channel.queueDeclare(qName, true, false, false, null);
        } catch (IOException e) {
            throw new VmsClientException(e.toString(), e);
        }
    }
}
